package com.dz.time;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import time.stock.StockPrice;
import time.stock.StockSource;

import java.util.concurrent.TimeUnit;

/**平均函数
 * @ClassName MyAggregateFunction
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/29 11:20
 * @Version 1.0
 */
public class MyAggregateFunction
{

    //源文件的路径
    private  final  static  String  sourcePath ="D:\\git_code\\hatchflink\\flink-advance\\src\\main\\resources\\stock\\stock-tick-20200108.csv";
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment  environment = StreamExecutionEnvironment.getExecutionEnvironment();

        //默认就是processTime
        environment.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        DataStream<StockPrice> stockStream = environment.addSource(new StockSource(sourcePath));

        //根据股票代号进行分组 10秒统计
        //接受参数 INPUT: 输入对象
        //ACC  计算值  Tuple3(String, Double, Int) - (symbol, sum, count)
        //OUT  输出值 Tuple2(String,Double)-(symbol,aggre)
        stockStream.keyBy(s->s.symbol).timeWindow(Time.of(10, TimeUnit.SECONDS)).aggregate(new AggregateFunction<StockPrice, Tuple3<String,Double,Integer>, Tuple2<String,Double>>() {


            //创建一个新的累加器
            @Override
            public Tuple3<String, Double, Integer> createAccumulator() {
                return Tuple3.of("",0D,0);
            }

            // 每条记录与迭代数据计算
            @Override
            public Tuple3<String, Double, Integer> add(StockPrice value, Tuple3<String, Double, Integer> accumulator) {

                //进行相加
                Double price = value.price+accumulator.f1;
                Integer sum  =  accumulator.f2+1;

                return Tuple3.of(value.symbol,price,sum);
            }

            @Override
            public Tuple2<String, Double> getResult(Tuple3<String, Double, Integer> accumulator) {
                return Tuple2.of(accumulator.f0,accumulator.f1/accumulator.f2);
            }

            //合并两个累加器，返回一个具有合并状态的累加器
            @Override
            public Tuple3<String, Double, Integer> merge(Tuple3<String, Double, Integer> a, Tuple3<String, Double, Integer> b) {
                return Tuple3.of(a.f0,a.f1+b.f1,a.f2+b.f2);
            }
        }).print();

        environment.execute("aggre job");
    }
}
